package com.kingsoft.minibase;


import java.io.IOException;
import java.util.Comparator;
import java.util.Optional;

/**
 * 用于构造KeyValue对象
 * 1. 属性值，KeyValue中的字段
 * 1. 枚举类，区分put和delete
 * 2. 构造KeyValue对象，私有化构造方法
 * 3. 对各字段进行编码，转化为字节数组
 * 4. 对字节数组进行解码
 * 4. 比较器，compareTo和equals
 */
public class KeyValue implements Comparable<KeyValue> {
    private static final int RAW_KEY_LEN_SIZE = 4;
    private static final int VAL_LEN_SIZE = 4;
    private static final int Op_SIZE = 1;
    private static final int SEQ_ID_SIZE = 8;

    // 注意属性值op使用小写，避免与枚举类名Op冲突
    private Op op;
    private byte[] key;
    private byte[] value;
    private long sequenceId;


    public enum Op {
        Put((byte) 0),
        Delete((byte) 1);

        private byte code;

        Op(byte code) {
            this.code = code;
        }

        public static Op code2Op(byte code) {
            switch (code) {
                case 0:
                    return Put;
                case 1:
                    return Delete;
                default:
                    throw new IllegalArgumentException("Unknown code: " + code);
            }
        }

        public byte getCode() {
            return this.code;
        }
    }
    
    private KeyValue(byte[] key, byte[] value, Op op, long sequenceId) {
        assert key != null;
        assert value != null;
        assert op != null;
        assert sequenceId >= 0;
        this.key = key;
        this.value = value;
        this.op = op;
        this.sequenceId = sequenceId;
    }

    public static KeyValue create(byte[] key, byte[] value, Op op, long sequenceId) {
        return new KeyValue(key, value, op, sequenceId);
    }

    public static KeyValue createPut(byte[] key, byte[] value, long sequenceId) {
        return KeyValue.create(key, value, Op.Put, sequenceId);
    }

    public static KeyValue createDelete(byte[] key, long sequenceId) {
        return KeyValue.create(key, Bytes.EMPTY_BYTES, Op.Delete, sequenceId);
    }


    public byte[] getKey() {
        return key;
    }

    public byte[] getValue() {
        return value;
    }

    // code用于序列化使用
    public Op getOp() {
        return this.op;
    }

    public long getSequenceId() {
        return this.sequenceId;
    }

    // rawKey的长度
    private int getRawKeyLen() {
        return key.length + Op_SIZE + SEQ_ID_SIZE;
    }

    //对KeyValue编码
    public byte[] toBytes() {
        int pos = 0;
        int rawKeyLen = getRawKeyLen();
        byte[] bytes = new byte[getSerializeSize()];

        // encode rawKeyLen
        byte[] rawKeyLenBytes = Bytes.toBytes(rawKeyLen);
        // 0和pos都是起始位置
        System.arraycopy(rawKeyLenBytes, 0, bytes, pos, rawKeyLen);
        pos += rawKeyLen;

        // encode value len
        byte[] valueLenBytes = Bytes.toBytes(value.length);
        System.arraycopy(valueLenBytes, 0, bytes, pos, value.length);
        pos += value.length;

        // encode rawKey:  key + op + sequenceId
        System.arraycopy(key, 0, bytes, pos, key.length);
        pos += key.length;
        byte[] opBytes = Bytes.toBytes(op.getCode());
        System.arraycopy(opBytes, 0, bytes, pos, 1);
        pos += 1;
        byte[] sequenceIdBytes = Bytes.toBytes(sequenceId);
        System.arraycopy(sequenceIdBytes, 0, bytes, pos, sequenceIdBytes.length);
        pos += sequenceIdBytes.length;

        // encode value
        System.arraycopy(value, 0, bytes, pos, value.length);

        return bytes;
    }

    public static KeyValue parseFrom(byte[] bytes, int offset) throws IOException {
        if (bytes == null) {
            throw new IOException("buffer is null");
        }
        if (offset + RAW_KEY_LEN_SIZE + VAL_LEN_SIZE >= bytes.length) {
            throw new IOException("Invalid offset or len. offset: " + offset + ", len: " + bytes.length);
        }
        // 先获取rawKeyLen，然后减去Op_SIZE和SEQ_ID_SIZE获取key的长度
        int pos = offset;
        int rawKeyLen = Bytes.toInt(Bytes.slice(bytes, pos, RAW_KEY_LEN_SIZE));
        pos += RAW_KEY_LEN_SIZE;
        //key所占用的长度
        int keyLen = rawKeyLen - Op_SIZE - VAL_LEN_SIZE;
        int valueLen = Bytes.toInt(Bytes.slice(bytes, pos, VAL_LEN_SIZE));
        pos += VAL_LEN_SIZE;

        // Decode key
        byte[] key = Bytes.slice(bytes, pos, keyLen);
        pos += key.length;

        // Decode op
        Op op = Op.code2Op(bytes[pos]);
        pos += 1;

        // Decode sequenceId
        long sequenceId = Bytes.toLong(Bytes.slice(bytes, pos, SEQ_ID_SIZE));
        pos += SEQ_ID_SIZE;

        // Decode value
        byte[] value = Bytes.slice(bytes, pos, valueLen);

        return create(key, value, op, sequenceId);
    }

    public static KeyValue parseFrom(byte[] bytes) throws IOException {
        return parseFrom(bytes, 0);
    }

    public int getSerializeSize() {
        return RAW_KEY_LEN_SIZE + VAL_LEN_SIZE + getRawKeyLen() + value.length;
    }

    // 对rowKey进行比较
    // 被排序的类需要实现Comparable接口，在使用Arrays.sort 或 Collections.sort排序方法时，会自动调用compareTo方法
    @Override
    public int compareTo(KeyValue keyValue) {
        if (keyValue == null) {
            throw new IllegalArgumentException("kv to compare should be null");
        }
        int ret = Bytes.compare(this.key, keyValue.key);
        if (ret != 0) {
            return ret;
        }

        if (this.sequenceId != keyValue.sequenceId) {
            return this.sequenceId > keyValue.sequenceId ? -1 : 1;
        }

        if (this.op != keyValue.op) {
            return this.op.getCode() > keyValue.op.getCode() ? -1 : 1;
        }
        return 0;
    }

    @Override
    public boolean equals(Object kv) {
        if (kv == null) {
            return false;
        }
        if (!(kv instanceof KeyValue)) {
            return false;
        }
        KeyValue that = (KeyValue) kv;
        return compareTo(that) == 0;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("key=").append(Bytes.toHex(this.key)).append("/op=").append(op).append
                ("/sequenceId=").append(this.sequenceId).append("/value=").append(Bytes.toHex(this
                .value));
        return sb.toString();
    }

    // Comparable接口必须去实现，排序逻辑是固定的
    // 而Comparator可以在不改变类本身的情况下，去定义多种排序方式
    private static class KeyValueComparator implements Comparator<KeyValue> {
        @Override
        public int compare(KeyValue a, KeyValue b) {
            if (a == b) return 0;
            if (a == null) return -1;
            if (b == null) return 1;
            // 只要key一样，值就是一样的，只用比较key
            return a.compareTo(b);
        }
    }

}
